package com.controller;

import lombok.SneakyThrows;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @Description:
 * @Auther: liuyanhui
 * @Date: 2022/11/21/ 14:14
 */
public class CompletableFutureDemo {


    // 异步执行无返回
    private static void testRunAsync() {
        ExecutorService executor = Executors.newFixedThreadPool(20);
        // 多线程处理
        CompletableFuture.runAsync(() -> {
            System.out.println("1 + 3 = " + (1 + 3));
        }, executor);
    }

    // 异步执行有返回
    private static void testSupplyAsync() throws ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(20);
        // 多线程处理
        CompletableFuture<Integer> res1 = CompletableFuture.supplyAsync(() -> {
            return 1 + 3;
        }, executor);
        System.out.println("res1 = " + res1.get());
    }


    // 结果处理  whenComplete   exceptionally
    @SneakyThrows
    private static void testWhenComplete() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
            }
            if (new Random().nextInt(10) % 2 == 0) {
                int i = 12 / 0;
            }
            System.out.println("执行结束！");
            return "test";
        });
        // 任务完成或异常方法完成时执行该方法
        // 如果出现了异常,任务结果为null
        future.whenComplete(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String t, Throwable action) {
                System.out.println(t + " 执行完成！");
            }
        });
        // 出现异常时先执行该方法
        future.exceptionally(new Function<Throwable, String>() {
            @Override
            public String apply(Throwable t) {
                System.out.println("执行失败：" + t.getMessage());
                return "异常xxxx";
            }
        });
//        future.get();
        future.join();
    }

    // 依赖关系 结果转换
    private static void testThenApply() {
        // 计算 （1+3） * （5+6）
        CompletableFuture<Integer> future = CompletableFuture
                .supplyAsync(() -> 1 + 3)
                .thenApply(number -> number * (5 + 6));
        Integer resu = future.join();
        System.out.println("resu = " + resu);
    }


    // 依赖关系 结果转换
    private static void testThenCompose() {
        // thenApply 和 thenCompose的区别：
        //thenApply转换的是泛型中的类型，返回的是同一个CompletableFuture；
        //thenCompose将内部的CompletableFuture调用展开来并使用上一个CompletableFutre调用的结果在下一步的CompletableFuture调用中进行运算，是生成一个新的CompletableFuture。
        CompletableFuture<Integer> future = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(30);
                        System.out.println("第一次运算：" + number);
                        return number;
                    }
                })
                .thenCompose(new Function<Integer, CompletionStage<Integer>>() {
                    @Override
                    public CompletionStage<Integer> apply(Integer param) {
                        return CompletableFuture.supplyAsync(new Supplier<Integer>() {
                            @Override
                            public Integer get() {
                                int number = param * 2;
                                System.out.println("第二次运算：" + number);
                                return number;
                            }
                        });
                    }
                });
        future.join();
    }

    // 结果消费
    // thenAccept()：对单个结果进行消费
    // thenAcceptBoth()：对两个结果进行消费
    // thenRun()：不关心结果，只对结果执行Action
    private static void testThenAccept() {
        // 计算 （1+3） * （5+6）
        CompletableFuture
                .supplyAsync(() -> 1 + 3)
                .thenAccept(number -> System.out.println("number = " + (number * (5 + 6))));
    }

    // henAcceptBoth()：对两个结果进行消费
    private static void testThenAcceptBoth() {
        CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                return 1 + 3;
            }
        });

        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                return 5 + 6;
            }
        });

        futrue1.thenAcceptBoth(future2, new BiConsumer<Integer, Integer>() {
            @Override
            public void accept(Integer x, Integer y) {
                System.out.println("最终结果：" + (x * y));
            }
        });

    }

    // thenRun也是对线程任务结果的一种消费函数，与thenAccept不同的是，thenRun会在上一阶段 CompletableFuture计算完成的时候执行一个Runnable，而Runnable并不使用该CompletableFuture计算的结果。
    private static void testThenRun() {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            int i = 1 + 3;
            System.out.println("i = " + i);
            return i;
        }).thenRun(() ->
                System.out.println("thenRun 执行"));
        future.join();

    }

    // 任务交互
    private static void testApplyToEither() {
        // 两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的转化操作。
        CompletableFuture<Integer> future1 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = 1 + 3;
                        System.out.println("任务1结果:" + number);
                        return number;
                    }
                });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = 5 + 6;
                System.out.println("任务2结果:" + number);
                return number;
            }
        });

        future1.applyToEither(future2, new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer number) {
                System.out.println("最快结果：" + number);
                return number;
            }
        });

    }

    // 两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的消费操作。
    private static void testAcceptEither() {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = 1 + 3;
                System.out.println("第一阶段：" + number);
                return number;
            }
        });

        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = 5 + 6;
                System.out.println("第二阶段：" + number);
                return number;
            }
        });

        CompletableFuture<Void> voidCompletableFuture = future1.acceptEither(future2, new Consumer<Integer>() {
            @Override
            public void accept(Integer number) {
                System.out.println("最快结果：" + number);
            }
        });
        voidCompletableFuture.join();

    }

    // 两个线程任务相比较，有任何一个执行完成，就进行下一步操作，不关心运行结果。
    private static void testRunAfterEither() {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = 1 + 3;
                System.out.println("任务1结果：" + number);
                return number;
            }
        });

        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = 5 + 6;
                System.out.println("任务2结果:" + number);
                return number;
            }
        });

        future1.runAfterEither(future2, new Runnable() {
            @Override
            public void run() {
                System.out.println("已经有一个任务完成了");
            }
        }).join();

    }

    // anyOf() 的参数是多个给定的 CompletableFuture，当其中的任何一个完成时，方法返回这个 CompletableFuture。
    private static void testAnyOf() {
        Random random = new Random();
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(random.nextInt(5));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(random.nextInt(1));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "world";
        });
        CompletableFuture<Object> result = CompletableFuture.anyOf(future1, future2);
        System.out.println("result = " + result.join());
    }

    // allOf方法用来实现多 CompletableFuture 的同时返回。
    private static void testAllOf() {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future1完成！");
            return "future1完成！";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("future2完成！");
            return "future2完成！";
        });
        CompletableFuture.allOf(future1, future2).join();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {

//        testCompletableFuture();
        // 异步执行无返回
//        testRunAsync();
        // 异步执行又返回,get 获取结果抛异常
//        testSupplyAsync();
        // 异步执行又返回,join 获取结果不抛异常
//        testSupplyAsync_join();
        // 结果处理  whenComplete（结果完成）   exceptionally（出现异常）
//        testWhenComplete();
        //thenApply转换的是泛型中的类型，返回的是同一个CompletableFuture；
//        testThenApply();
        //thenCompose将内部的CompletableFuture调用展开来并使用上一个CompletableFutre调用的结果在下一步的CompletableFuture调用中进行运算，是生成一个新的CompletableFuture。
//        testThenCompose();
        // 结果消费
//        testThenAccept();
        // 对两个结果进行消费
//        testThenAcceptBoth();
        // thenRun也是对线程任务结果的一种消费函数，与thenAccept不同的是，thenRun会在上一阶段 CompletableFuture计算完成的时候执行一个Runnable，而Runnable并不使用该CompletableFuture计算的结果。
//        testThenRun();
        // 两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的转化操作。
//        testApplyToEither();
        // 两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的消费操作。
//        testAcceptEither();
        // 两个线程任务相比较，有任何一个执行完成，就进行下一步操作，不关心运行结果。
//        testRunAfterEither();
        // anyOf() 的参数是多个给定的 CompletableFuture，当其中的任何一个完成时，方法返回这个 CompletableFuture。
//        testAnyOf();
        // allOf方法用来实现多 CompletableFuture 的同时返回。
//        testAllOf();
        testFuture();
    }

    // 异步执行又返回,join 获取结果不抛异常
    private static void testSupplyAsync_join() {
        ExecutorService executor = Executors.newFixedThreadPool(20);
        // 多线程处理
        CompletableFuture<Integer> res1 = CompletableFuture.supplyAsync(() -> {
            return 1 + 3;
        }, executor);
        System.out.println("res1 = " + res1.join());
    }


    private static void testCompletableFuture() {
        ExecutorService executor = Executors.newFixedThreadPool(20);
        // 多线程处理
        List<CompletableFuture> futures = new ArrayList<>();
        int totalPage = 10;
        System.out.println("开始执行！");
        for (int i = 0; i < totalPage; i++) {
            futures.add(CompletableFuture.runAsync(() -> {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread());
            }, executor));
        }
        CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new)).join();
        System.out.println("执行完成！");
    }

    // 练习,模拟所线程查询，最后统计查询结果
    private static void testFuture() {
        ExecutorService executor = Executors.newFixedThreadPool(20);
        // 多线程处理
        List<CompletableFuture> futures = new ArrayList<>();
        int totalPage = 10;
        System.out.println("开始执行！");
        List<String> resultList = new ArrayList<>();
        List<String> errorList = new ArrayList<>();
        for (int i = 0; i < totalPage; i++) {
            futures.add(CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread());
                return Thread.currentThread().getName();
            }, executor).handle((result, throwable) -> {
                if (null == throwable) {
                    resultList.add(result);
                } else {
                    errorList.add(throwable.getMessage());
                }
                return Thread.currentThread() + "--完成";
            }));
        }
        CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new)).whenComplete((v, th) -> {
            System.out.println("查询完成  = ");
            System.out.println(resultList);
            System.out.println(errorList);
        }).join();
        System.out.println("执行完成！");
    }

}
